package skewreduce.framework.physical;

import java.io.IOException;

import org.apache.hadoop.mapreduce.JobID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.SchedulerEvent;


/**
 * physical instantiation of a logical operator.
 */
public abstract class BlockingPhysicalOp
extends PhysicalOp {
    private static Logger LOG = LoggerFactory.getLogger("PLAN");

    private boolean waitForCompletion()
    throws IOException, InterruptedException {
        while ( ! job.isComplete() ) {
            Thread.sleep(5000);
        }
        return job.isSuccessful();
    }
    
    public SchedulerEvent call() throws Exception {
        SchedulerEvent result = null;

        if ( LOG.isInfoEnabled() ) {
        	LOG.info("EXECUTE {}",getID());
        }

        try {
        	setup();
        	
        	scheduledAt = System.currentTimeMillis();

            job.submit();

            //plan.registerRunningPhysicalOp(this);

            setStatus(Status.RUNNING);

            JobID jobid = job.getJobID();

            if ( LOG.isInfoEnabled() ) {
                LOG.info("launched {} as {}",getID(),jobid);
            }

            // waitForCompletion does not respect interrupted exception
            //boolean ok = job.waitForCompletion(false);
            boolean ok = waitForCompletion();
            if ( ok ) {
            	//result = new SchedulerEvent.Completion(this);
            	//setStatus(Status.COMPLETE);
            } else {
            	//result = new SchedulerEvent.Failed(this);
            	//setStatus(Status.FAILED);
            }
        } catch ( InterruptedException x ) {
            LOG.info("Killing MapReduce task : "+job.getJobID());
            try { job.killJob(); } catch ( IOException ignore ) {}
            return new SchedulerEvent.Killed(this);
            //} else {
            //    result = new SchedulerEvent.Error(this,x);
            //}
        } catch ( Exception x ) {
            return new SchedulerEvent.Error(this,x);
        }/* finally {
        	switch ( result.getEventType() ) {
        	case COMPLETE:
        		onSuccess();
        	default:
        		onFail();
        		break;
        	}
        	cleanup();
        }*/
        
        //return result;
        return new SchedulerEvent.Noop(this);
    }
}
